package com.maxt.common.mqtt;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.util.StringUtils;

/**
 * @autor create by jerry
 * @date 2017年5月21日 下午4:06:32
 **/
public class Subscriber {
    public static final String HOST = "tcp://ext.prod.emqtt1.iwangfan.cn:1883"; // 设置服务器地址,mqtt使用的是TCP/IP连接

    public static String SUBSCRIBE_TOPIC_NAME_1 = "tiger";
    // public static String SUBSCRIBE_TOPIC_NAME_2 = "tiger/flow/GD200A15A003268/#";
    // public static String SUBSCRIBE_TOPIC_NAME_3 = "dma/flow/GD200A15A003268/#";
    // public static String SUBSCRIBE_TOPIC_NAME_4 = "tiger/flow/GD200A15A003059/#";
    // public static String SUBSCRIBE_TOPIC_NAME_5 = "dma/flow/GD200A15A003059/#";
    //public static String SUBSCRIBE_TOPIC_NAME_2 = "tiger/flow/#";
    public static String SUBSCRIBE_TOPIC_NAME_3 = "dma/flow/#";
    //public static String SUBSCRIBE_TOPIC_NAME_4 = "tiger/flow/HMAPA01160700537/#";
    //public static String SUBSCRIBE_TOPIC_NAME_5 = "dma/flow/HMAPA01160701286/#";
    public static String SUBSCRIBER_ID = "HM_MAXT"; // 订阅者ID

    private MqttClient subscriber;
    private MqttConnectOptions options;

    private final String USERNAME = "admin";
    private final String PASSWORD = "public";

    private ScheduledExecutorService scheduler;// 定时执行某个任务

    public void reconnect() {
        scheduler = Executors.newSingleThreadScheduledExecutor();// 生成只有一个线程的线程池
        scheduler.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                if (!subscriber.isConnected()) {
                    try {
                        subscriber.connect();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }, 0, 10 * 1000, TimeUnit.MILLISECONDS);// 每10毫秒重试执行一次,0秒延时
    }

    public void start() {
        try {
            // host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            subscriber = new MqttClient(HOST, SUBSCRIBER_ID, new MemoryPersistence());
            // MQTT的连接设置
            options = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(true);
            // 设置连接的用户名
            options.setUserName(USERNAME);
            // 设置连接的密码
            options.setPassword(PASSWORD.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(10);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(20);
            // 设置回调
            subscriber.setCallback(new PushCallBack());
            // MqttTopic topic = subscriber.getTopic(SUBSCRIBE_TOPIC_NAME_1);
            // setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
            // options.setWill(topic, "close".getBytes(), 0, true);

            subscriber.connect(options);
            // 订阅消息
            int[] Qos = { 1, 1  };
            // String[] topic1 =
            // {SUBSCRIBE_TOPIC_NAME_1,SUBSCRIBE_TOPIC_NAME_2,SUBSCRIBE_TOPIC_NAME_3,SUBSCRIBE_TOPIC_NAME_4,SUBSCRIBE_TOPIC_NAME_5};
            String[] topic1 = {"cool_999","cool_one"};//{ SUBSCRIBE_TOPIC_NAME_1,  SUBSCRIBE_TOPIC_NAME_3 };
            subscriber.subscribe(topic1, Qos);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws MqttException {
        Subscriber subscriber = new Subscriber();
        subscriber.start();
    }

    public class PushCallBack implements MqttCallback {

        /**
         * 丢失连接,重连
         */
        @Override
        public void connectionLost(Throwable cause) {
            System.out.println("连接丢失");
            if (subscriber.isConnected())
                reconnect();
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            System.out.println("deliveryComplete---------" + token.isComplete());
        }

        /**
         * 向指定的文件的最后，追加指定的内容
         * 
         * @param mainFile
         * @param appendText
         * @return
         */
        public boolean appendTextToFile(String mainFile, String appendText) {
            if (StringUtils.isEmpty(appendText))
                return false;

            File main = new File(mainFile);
            if (!main.exists()) {
                if (main.getParentFile() != null)
                    main.getParentFile().mkdirs(); // 确保上级目录已创建
                try {
                    main.createNewFile(); // 创建文件
                } catch (IOException e) {
                }
            }

            // 构造函数中的第二个参数true表示以追加形式写文件
            try (FileWriter writer = new FileWriter(mainFile, true)) {
                writer.write("\n");
                writer.write(appendText);
                writer.close();
            } catch (Exception e) {
                return false;
            }
            return true;
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            // TODO Auto-generated method stub
            // subscribe后得到的消息会执行到这里面

            System.out.println("接收消息主题 : " + topic);
            System.out.println("接收消息Qos : " + message.getQos());
            System.out.println("接收消息内容 : " + new String(message.getPayload()));
            System.out.println("第" + message.getId() + "条");

//            appendTextToFile("d:\\temp\\wangfan\\mqtt_topic.log", topic);
            // appendTextToFile("d:\\temp\\wangfan\\mqtt.log",topic);
            // appendTextToFile("d:\\temp\\wangfan\\mqtt.log",new String(message.getPayload()));

            Map<String, String> map = new HashMap<String, String>();

            String[] arr = topic.split("/");
            if (arr != null && arr.length == 4 && !arr[3].equals("traffic") && !arr[3].equals("onInternet")) {
                if (map.keySet() != null && !map.keySet().contains(arr[2])) {
                    map.put(arr[2], arr[3]);
//                    appendTextToFile("d:\\temp\\wangfan\\mqtt_topic_single.log", topic);
                    System.out.println(topic);
                }
            }

        }

    }
}
